package org.apache.ocean.main.server;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.URL;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.SortedSet;
import java.util.concurrent.ExecutorService;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.BooleanUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.lucene.index.Term;
import org.apache.lucene.queryParser.ParseException;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.TermQuery;
import org.apache.lucene.search.Weight;
import org.apache.ocean.ClusterConstants;
import org.apache.ocean.main.Deletes;
import org.apache.ocean.main.DiskIndex;
import org.apache.ocean.main.IndexDocuments;
import org.apache.ocean.main.IndexSchema;
import org.apache.ocean.main.Snapshot;
import org.apache.ocean.main.TransactionMonitor;
import org.apache.ocean.main.TransactionSystem;
import org.apache.ocean.main.Batch.MasterBatch;
import org.apache.ocean.main.Deletes.DeleteByQuery;
import org.apache.ocean.main.DiskIndex.DiskIndexSnapshot;
import org.apache.ocean.main.IndexDocuments.IndexDocument;
import org.apache.ocean.main.TransactionSystem.Category;
import org.apache.ocean.main.TransactionSystem.Category.Indices;
import org.apache.ocean.main.client.OceanXML;
import org.apache.ocean.main.client.SearchClient;
import org.apache.ocean.main.http.HttpParameters;
import org.apache.ocean.main.replication.ByteBufferPool;
import org.apache.ocean.main.replication.Replication;
import org.apache.ocean.main.replication.ByteBufferPool.ByteBuffer;
import org.apache.ocean.main.replication.ByteBufferPool.ByteBuffers;
import org.apache.ocean.main.replication.LogFile2.Record;
import org.apache.ocean.main.replication.RawLogFile.Data;
import org.apache.ocean.main.replication.RawLogFile.Load;
import org.apache.ocean.main.replication.RawLogFile.RecordHeader;
import org.apache.ocean.main.replication.Replication.Checksum;
import org.apache.ocean.main.replication.Replication.Node;
import org.apache.ocean.main.replication.Replication.TransactionParticipant;
import org.apache.ocean.main.search.Filters;
import org.apache.ocean.main.search.OceanQueryParser;
import org.apache.ocean.main.search.Search;
import org.apache.ocean.main.search.SnapshotSearch;
import org.apache.ocean.main.search.TermFrequencies;
import org.apache.ocean.main.search.DocumentSearch.DocumentQuery;
import org.apache.ocean.main.search.DocumentSearch.DocumentResult;
import org.apache.ocean.main.search.Filters.QueryFilter;
import org.apache.ocean.main.search.Search.FieldDoc;
import org.apache.ocean.main.search.Search.Result;
import org.apache.ocean.main.search.Search.ScoreDoc;
import org.apache.ocean.main.search.Search.SearchQuery;
import org.apache.ocean.main.search.Search.Sort;
import org.apache.ocean.main.search.cache.FieldCacheKey.FieldIndex;
import org.apache.ocean.main.util.Constants;
import org.apache.ocean.main.util.PrintStringWriter;
import org.apache.ocean.main.util.Timeout;
import org.apache.ocean.main.util.Util;
import org.apache.ocean.main.util.XMLUtil;
import org.jdom.Element;

public class TransactionServlet extends HttpServlet {
	public static Logger log = Logger.getLogger(TransactionServlet.class.getName());
	private TransactionSystem transactionSystem;
	private TransactionServer transactionServer;

	private OceanXML defaultOceanResponse() throws Exception {
		OceanXML oceanResponse = new OceanXML();
		oceanResponse.addHeader("serverid", transactionServer.getId());
		oceanResponse.addHeader("serverurl", transactionServer.getUrl());
		oceanResponse.addHeader("servercell", transactionServer.getCell());
		return oceanResponse;
	}

	public void service(HttpServletRequest servletRequest, HttpServletResponse servletResponse) throws ServletException, IOException {
		HttpParameters httpParameters = Util.toHttpParameters(servletRequest);
		if (httpParameters.has("action", "gettransactions")) {
			handleGetTransactions(httpParameters, servletResponse);
			return;
		} else if (httpParameters.has("action", "getfile")) {
			handleGetFile(httpParameters, servletResponse);
			return;
		}
		PrintStringWriter printStringWriter = new PrintStringWriter();
		Response response = new Response(printStringWriter);
		try {
			if (StringUtils.equalsIgnoreCase("post", servletRequest.getMethod())) {
				OceanXML oceanResponse = defaultOceanResponse();
				if (httpParameters.has("action", "writetransaction")) {
					handleWriteTransaction(servletRequest, oceanResponse);
				} else {
					OceanXML oceanRequest = OceanXML.parse(servletRequest.getReader());
					if (httpParameters.has("action", "writebatch")) {
						handleWriteBatch(oceanRequest, oceanResponse);
					} else if (httpParameters.has("action", "gettransactiondifferences")) {
						handleGetTransactionDifferences(oceanRequest, oceanResponse);
					}
				}
				response.contentType = ClusterConstants.CONTENT_TYPE_XML;
				Element responseElement = oceanResponse.toElement();
				XMLUtil.outputElement(responseElement, printStringWriter);
			} else if (StringUtils.equalsIgnoreCase("get", servletRequest.getMethod())) {
				if (httpParameters.has("action", "transactionprocess")) {
					handleTransactionProcess(httpParameters, servletResponse);
				} else {
					OceanXML oceanResponse = defaultOceanResponse();
					String action = httpParameters.get("action");
					if (StringUtils.isBlank(action)) {
						handleSearch(httpParameters, oceanResponse);
					} else if (StringUtils.equals("getsnapshotfiles", action)) {
						handleSnapshotGetFiles(httpParameters, oceanResponse);
					} else if (StringUtils.equals("getnode", action)) {
						handleGetNode(httpParameters, oceanResponse);
					} else if (StringUtils.equals("getchosenmaster", action)) {
						handleGetChosenMaster(httpParameters, oceanResponse);
					} else if (StringUtils.equals("renewlease", action)) {
						handleRenewLease(httpParameters, oceanResponse);
					} else if (StringUtils.equals("gettermfrequencies", action)) {
						handleTermFrequencies(httpParameters, oceanResponse);
					} else if (StringUtils.equals("getdocuments", action)) {
						handleGetDocuments(httpParameters, oceanResponse);
					} else if (StringUtils.equals("searchdocids", action)) {
						handleSearchDocIds(httpParameters, oceanResponse);
					} else if (StringUtils.equals("getchecksum", action)) {
						handleGetChecksum(httpParameters, oceanResponse);
					} else if (StringUtils.equals("getcategorychecksum", action)) {
						handleGetCategoryChecksum(httpParameters, oceanResponse);
					} 
				}
			}
			servletResponse.setStatus(200);
		} catch (Throwable throwable) {
			try {
				log.log(Level.SEVERE, "", throwable);
				servletResponse.setStatus(500);
				String responseXML = OceanXML.toOceanXMLString(throwable);
				printStringWriter = new PrintStringWriter();
				printStringWriter.print(responseXML);
			} catch (Throwable exTh) {
				log.log(Level.SEVERE, "", throwable);
				log.log(Level.SEVERE, "inside creating exception response", exTh);
				throw new ServletException(exTh);
			}
		} finally {
			if (printStringWriter != null) {
				servletResponse.setContentType(response.contentType);
				PrintWriter writer = servletResponse.getWriter();
				writer.write(printStringWriter.toString());
			}
		}
	}
  
	private void handleGetCategoryChecksum(HttpParameters parameters, OceanXML oceanResponse) throws Exception {
		Replication replication = transactionSystem.getReplication();
		Long highId = parameters.getLong("highid");
		String category = parameters.get("category");
		Checksum checksum = replication.getCheckSum(category, highId);
		oceanResponse.addBody(checksum);
	}
	
	private void handleGetChecksum(HttpParameters parameters, OceanXML oceanResponse) throws Exception {
		Replication replication = transactionSystem.getReplication();
		Long highId = parameters.getLong("highid");
		Checksum checksum = replication.getCheckSum(highId);
		oceanResponse.addBody(checksum);
	}
	
	private void handleGetDocuments(HttpParameters parameters, OceanXML oceanResponse) throws Exception {
		String category = parameters.get("category");
		Long snapshotid = parameters.getLong("snapshotid");
		String fieldsStr = parameters.get("fields");
		SortedSet<String> fields = Util.splitToSortedSet(fieldsStr);
		List<Long> docIds = new ArrayList<Long>();
		for (String docidStr : parameters.getValues("docid")) {
			docIds.add(new Long(docidStr));
		}
		IndexDocuments indexDocuments = transactionServer.getGetDocuments(category, snapshotid, docIds, fields);
		for (IndexDocument indexDocument : indexDocuments) {
		  oceanResponse.addBody(indexDocument);
		}
	}
	
	private void handleGetTransactionDifferences(OceanXML oceanRequest, OceanXML oceanResponse) throws Exception {
		Replication replication = transactionSystem.getReplication();
		List<Long> differences = new ArrayList<Long>();
		for (Element element : oceanRequest.getBodyChildren()) {
			Long id = new Long(element.getText());
			if (!replication.contains(id)) {
				differences.add(id);
			}
		}
		for (Long id : differences) {
			Element idElement = new Element("id");
			idElement.setText(id.toString());
		  oceanResponse.addBody(idElement);
		}
	}
	
	private void handleTermFrequencies(HttpParameters parameters, OceanXML oceanResponse) throws Exception {
		String queryString = parameters.get("q");
		String categoryString = parameters.get("category");
		long schemaVersion = parameters.getLong("schemaversion");
		Category category = transactionSystem.getCategory(categoryString);
		Snapshot snapshot = category.getSnapshots().getLatestSnapshot();
		IndexSchema.Category indexSchemaCategory = category.getCategoryIndexSchema(schemaVersion);
		Query query = OceanQueryParser.parse(queryString, indexSchemaCategory);
		Query rewrittenQuery = transactionServer.rewriteQuery(query, snapshot);
		TermFrequencies termFrequencies = transactionServer.getTermFrequencies(rewrittenQuery, snapshot);
		oceanResponse.addBody(termFrequencies);
	}
	
	private void handleTransactionProcess(HttpParameters parameters, HttpServletResponse servletResponse) throws Exception {
		String serverid = parameters.get("serverid");
		URL url = parameters.getUrl("url");
		Long snapshotId = parameters.getLong("id");
		TransactionParticipant transactionParticipant = new TransactionParticipant(serverid, url);
		long timeoutDuration = parameters.getLong("timeout");
		Timeout timeout = new Timeout(timeoutDuration);
		Replication replication = transactionSystem.getReplication();
		TransactionMonitor<Long, TransactionParticipant> transactionMonitor = replication.getTransactionMonitor();
		PrintWriter writer = servletResponse.getWriter();
		try {
			transactionMonitor.waitPrepare(snapshotId, transactionParticipant, timeout);
			writer.println("prepare");
			transactionMonitor.waitCommit(snapshotId, transactionParticipant, timeout);
			writer.println("commit");
		} catch (Exception exception) {
			writer.println("failed");
			log.log(Level.SEVERE, "", exception);
		}
	}
	
	public static class HttpSearchParameters {
		private final Query query;
		private final Sort sort;
    private final Filters filters;
    private Timeout timeout;
    private final int start;
    private final int num;
    private TermFrequencies termFrequencies;
    
    public HttpSearchParameters(HttpParameters parameters, IndexSchema.Category schemaCategory) throws ParseException {
    	String queryParameter = parameters.get("q");
    	String[] split = StringUtils.split(queryParameter, ";");
  		String queryString = split[0];
  		String sortString = null;
  		if (split.length > 1)
  			sortString = split[1];
  		else {
  			sortString = parameters.get("sort");
  		}
  		sort = Search.toSort(sortString);
  		num = parameters.getInt("num", 10);
  		start = parameters.getInt("start", 0);
  		query = OceanQueryParser.parse(queryString, schemaCategory);
  		Long timeoutValue = parameters.getLong("timeout");
  		if (timeoutValue != null) {
  			timeout = new Timeout(timeoutValue);
  		}
  		filters = new Filters();
  		for (String filterString : parameters.getValues("filter")) {
  			Query query = OceanQueryParser.parse(filterString, schemaCategory);
  			QueryFilter queryFilter = new QueryFilter(query);
  			filters.add(queryFilter);
  		}
  		termFrequencies = TermFrequencies.parse(parameters, schemaCategory);
    }
     
    public TermFrequencies getTermFrequencies() {
    	return termFrequencies;
    }
    
		public Query getQuery() {
			return query;
		}

		public Sort getSort() {
			return sort;
		}

		public Filters getFilters() {
			return filters;
		}

		public Timeout getTimeout() {
			return timeout;
		}

		public int getStart() {
			return start;
		}

		public int getNum() {
			return num;
		}
	}
	
	private void handleSearchDocIds(HttpParameters parameters, OceanXML response) throws Exception {
		String categoryName = parameters.get("category");
		Long snapshotId = parameters.getLong("snapshotid");
		Long schemaVersion = parameters.getLong("schemaversion");
		Category category = transactionSystem.getCategory(categoryName);
		IndexSchema.Category schemaCategory = category.getCategoryIndexSchema(schemaVersion);
		Snapshot snapshot = category.getSnapshots().get(snapshotId);
		HttpSearchParameters searchParameters = new HttpSearchParameters(parameters, schemaCategory);
		TermFrequencies termFrequencies = searchParameters.getTermFrequencies();
		Query query = searchParameters.getQuery();
		Weight weight = transactionServer.getWeight(query, termFrequencies, snapshot);
		int start = searchParameters.getStart();
		int num = searchParameters.getNum();
		Filters filters = searchParameters.getFilters();
		Sort sort = searchParameters.getSort();
		Timeout timeout = searchParameters.getTimeout();
		SearchQuery searchQuery = new SearchQuery(weight, start, num, filters, sort, timeout);
		ExecutorService threadPool = transactionServer.getSearchThreadPool();
		SnapshotSearch snapshotSearch = new SnapshotSearch(searchQuery, snapshot, threadPool);
		Result<ScoreDoc> result = snapshotSearch.call();
		response.addHeader("numfound", result.getNumFound());
		response.addHeader("snapshotid", snapshotId);
		IndexDocuments indexDocuments = new IndexDocuments();
		for (ScoreDoc scoreDoc : result.getDocs()) {
			FieldIndex fieldIndex = scoreDoc.indexSnapshot.getFieldIndex(Constants.ID);
			Long id = (Long)fieldIndex.valueIndices[scoreDoc.doc].value;
			IndexDocument indexDocument = new IndexDocument(schemaCategory);
			if (scoreDoc instanceof FieldDoc) {
				FieldDoc fieldDoc = (FieldDoc)scoreDoc;
				for (int x=0; x < sort.size(); x++) {  
					Object object = fieldDoc.comparables[x];
					Sort.Field sortField = sort.get(x);
					String name = sortField.getName();
					indexDocument.add(name, object);
				}
			}
			indexDocument.add(Constants.ID, id);
			indexDocument.add(Constants.SCORE, scoreDoc.score);
			// TODO: have schema handle score
			indexDocuments.add(indexDocument);
		}
		for (IndexDocument indexDocument : indexDocuments) {
			response.addBody(indexDocument);
		}
	}
	
	private void handleSearch(HttpParameters parameters, OceanXML response) throws Exception {
		String fieldsString = parameters.get("fields");
		String categoryName = parameters.get("category");
		Long schemaVersion = parameters.getLong("schemaversion");
		Category category = transactionSystem.getCategory(categoryName);
		
		IndexSchema.Category schemaCategory = category.getCategoryIndexSchema(schemaVersion);
		Snapshot snapshot = category.getSnapshots().getLatestSnapshot();
		HttpSearchParameters searchParameters = new HttpSearchParameters(parameters, schemaCategory);
		
		Weight weight = transactionServer.getWeight(searchParameters.getQuery(), snapshot);
		int start = searchParameters.getStart();
		int num = searchParameters.getNum();
		Filters filters = searchParameters.getFilters();
		SortedSet<String> fields = Util.splitToSortedSet(fieldsString);
		Sort sort = searchParameters.getSort();
		Timeout timeout = searchParameters.getTimeout();
		DocumentQuery documentQuery = new DocumentQuery(weight, start, num, filters, fields, sort, timeout);
		DocumentResult documentResult = transactionServer.search(documentQuery, snapshot);
		response.addHeader("numfound", documentResult.getNumFound());
		response.addHeader("snapshot", snapshot.getId());
		IndexDocuments indexDocuments = documentResult.getIndexDocuments();
		for (IndexDocument indexDocument : indexDocuments) {
			response.addBody(indexDocument.toElement());
		}
	}

	private void handleGetTransactions(HttpParameters parameters, HttpServletResponse servletResponse) {
		try {
			Long startid = parameters.getLong("startid");
			int num = parameters.getInt("num");
			Replication replication = transactionSystem.getReplication();
			EnumSet<Load> load = EnumSet.of(Load.DOCUMENTS, Load.DELETES);
			List<RecordHeader> recordHeaders = replication.getRecordHeaders(startid, num);
			servletResponse.setStatus(200);
			DataOutputStream dataOutput = new DataOutputStream(servletResponse.getOutputStream());
			dataOutput.writeInt(recordHeaders.size());
			for (RecordHeader recordHeader : recordHeaders) {
				dataOutput.writeLong(recordHeader.id);
				dataOutput.writeUTF(recordHeader.category);
				dataOutput.writeLong(recordHeader.schemaVersion);
				Record record = null;
				try {
					record = replication.getRecord(load, recordHeader);
					Data data = record.getData();
					ByteBuffers documentsByteBuffers = data.getDocumentsByteBuffers();
					dataOutput.writeInt(documentsByteBuffers.getLength());
					documentsByteBuffers.writeTo(dataOutput);
					ByteBuffers deletesByteBuffers = data.getDeletesByteBuffers();
					dataOutput.writeInt(deletesByteBuffers.getLength());
					deletesByteBuffers.writeTo(dataOutput);
				} finally {
					if (record != null)
						record.getData().finished();
				}
			}
		} catch (Throwable throwable) {
			servletResponse.setStatus(500);
			log.log(Level.SEVERE, "", throwable);
		}
	}

	private void handleRenewLease(HttpParameters parameters, OceanXML response) throws Exception {
		Replication replication = transactionSystem.getReplication();
		long duration = parameters.getLong("duration");
		Node node = new Node(parameters);
		boolean renewed = replication.renewLease(node, duration);
		response.addHeader("renewed", BooleanUtils.toStringTrueFalse(renewed));
	}

	private void handleWriteTransaction(HttpServletRequest servletRequest, OceanXML response) throws Exception {
		Replication replication = transactionSystem.getReplication();
		HttpParameters parameters = Util.toHttpParameters(servletRequest);
		Long id = parameters.getLong("id");
		Long previousId = parameters.getLong("previousid");
		String category = parameters.get("category");
		Long schemaVersion = parameters.getLong("schemaversion");
		ByteBufferPool documentByteBuffers = replication.getDocumentByteBuffers();
		ByteBufferPool deletesByteBuffers = replication.getDeletesByteBuffers();
		DataInputStream input = new DataInputStream(servletRequest.getInputStream());
		int documentsLength = input.readInt();
		ByteBuffer documentsBuffer = documentByteBuffers.get(documentsLength);
		int documentsRead = input.read(documentsBuffer.getBytes(), 0, documentsLength);
		if (documentsLength != documentsRead) {
			throw new Exception("documentsRead: " + documentsRead + " documentsLength: " + documentsLength);
		}
		int deletesLength = input.readInt();
		ByteBuffer deletesBuffer = deletesByteBuffers.get(deletesLength);
		int deletesRead = input.read(deletesBuffer.getBytes(), 0, deletesLength);
		if (deletesLength != deletesRead) {
			throw new Exception("deletesRead: " + deletesRead + " deletesLength: " + deletesLength);
		}
		Data data = new Data(documentsBuffer, documentsLength, deletesBuffer, deletesLength);
		if (replication.contains(previousId)) {
			replication.writeSlaveBatch(id, category, schemaVersion, data);
			response.addHeader("success", "true");
			response.addHeader("renewlease", "true");
			response.addHeader("leaseduration", Replication.LEASE_DURATION);
			Node node = replication.getNode();
			response.addBody(node);
		} else {
			response.addHeader("cancellease", "true");
			response.addHeader("success", "false");
		}
	}

	private void handleGetChosenMaster(HttpParameters parameters, OceanXML response) throws Exception {
		Replication replication = transactionSystem.getReplication();
		Node node = replication.getChosenMaster();
		response.addBody(node);
	}

	private void handleGetNode(HttpParameters parameters, OceanXML response) throws Exception {
		Replication replication = transactionSystem.getReplication();
		Node node = replication.getNode();
		response.addBody(node);
	}

	private void handleGetFile(HttpParameters parameters, HttpServletResponse servletResponse) {
		try {
			String category = parameters.get("category");
			Long indexid = parameters.getLong("indexid");
			String filename = parameters.get("filename");
			TransactionSystem.Category transactionSystemCategory = transactionSystem.getCategory(category);
			Indices indices = transactionSystemCategory.getIndices();
			DiskIndex diskIndex = (DiskIndex) indices.get(indexid);
			File file = diskIndex.getFile(filename);
			if (!file.exists()) {
				servletResponse.setStatus(404);
			} else {
				FileInputStream fileInput = new FileInputStream(file);
				OutputStream outputStream = servletResponse.getOutputStream();
				IOUtils.copy(fileInput, outputStream);
				fileInput.close();
			}
		} catch (IOException ioException) {
			servletResponse.setStatus(500);
		}
	}

	// TODO: finish handleSnapshotGetFiles
	private void handleSnapshotGetFiles(HttpParameters parameters, OceanXML oceanResponse) throws Exception {
		for (Category category : transactionSystem.getCategories()) {
			Snapshot snapshot = category.getSnapshots().getLatestSnapshot();
			List<DiskIndexSnapshot> diskIndexSnapshots = snapshot.getDiskIndexSnapshots();
			for (DiskIndexSnapshot diskIndexSnapshot : diskIndexSnapshots) {
				List<File> files = diskIndexSnapshot.getFiles();

			}
		}
	}

	private void handleWriteBatch(OceanXML oceanRequest, OceanXML oceanResponse) throws Exception {
		String categoryName = oceanRequest.getHeaderValue("category");
		Long schemaVersion = oceanRequest.getHeaderLong("schemaversion");
		Category category = transactionSystem.getCategory(categoryName);
		MasterBatch masterBatch = category.createMasterBatch(schemaVersion);
		IndexSchema.Category indexSchemaCategory = category.getCategoryIndexSchema(schemaVersion);
		SearchClient.Batch clientBatch = new SearchClient.Batch(oceanResponse.getFirstInBody(), indexSchemaCategory);
		IndexDocuments indexDocuments = clientBatch.getDocuments();
		Deletes deletes = new Deletes();
		List<String> deleteStrings = clientBatch.getDeletes();
		for (String deleteString : deleteStrings) {
			Query query = OceanQueryParser.parse(deleteString, indexSchemaCategory);
			if (query instanceof TermQuery) {
				TermQuery termQuery = (TermQuery) query;
				Term term = termQuery.getTerm();
				deletes.addTerm(term);
			} else {
				deletes.addDeleteByQuery(new DeleteByQuery(query));
			}
		}
		masterBatch.addDocuments(indexDocuments);
		masterBatch.setDeletes(deletes);
		masterBatch.commit();
	}

	public static class Response {
		public PrintStringWriter printStringWriter;
		public String contentType;

		public Response(PrintStringWriter printStringWriter) {
			this.printStringWriter = printStringWriter;
		}
	}
}
